package com.swak.netty.multicast;

import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;

import com.swak.Constants;
import com.swak.OS;
import com.swak.exception.RemotingException;
import com.swak.reactivex.threads.Contexts;
import com.swak.reactivex.transport.Channel;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangeHandlerAdapter;
import com.swak.reactivex.transport.ExchangePeer;
import com.swak.reactivex.transport.Exchangers;
import com.swak.reactivex.transport.TransportMode;
import com.swak.reactivex.transport.resources.LoopResources;
import com.swak.registry.URL;
import com.swak.utils.JsonMapper;
import com.swak.utils.Maps;

public class UdpPeer extends ExchangeHandlerAdapter {

	ExchangePeer peer;

	public UdpPeer(boolean useMulticast, int port) throws RemotingException {
		LoopResources resources = Contexts.createEventLoopResources(TransportMode.OS, 1, 2, "Test.EventLoop-", false, 2,
				TimeUnit.SECONDS);
		Map<String, String> parameters = Maps.newHashMap();
		parameters.put(Constants.Bind_Host, Constants.ANYHOST);
		parameters.put(Constants.Use_Udp, "true");
		if (useMulticast) {
			parameters.put(Constants.Use_Udp_Multicast, "true");
			parameters.put(Constants.Multicast_Port_Key, String.valueOf(port));
		}
		URL url = new URL("Test", OS.ip(), port, parameters);
		this.peer = Exchangers.peer(url, resources, this);
	}

	@Override
	public void connected(Channel channel) throws RemotingException {
		System.out.println("connected.");
	}

	@Override
	public void disconnected(Channel channel) throws RemotingException {
		System.out.println("disconnected.");
	}

	@Override
	public void sent(Channel channel, Object message) throws RemotingException {
		System.out.println("sent.");
	}

	@Override
	public void received(Channel channel, Object message) throws RemotingException {
		System.out.println("received：" + JsonMapper.toJson(message));
	}

	@Override
	public void caught(Channel channel, Throwable exception) throws RemotingException {
		System.out.println("caught.");
		exception.printStackTrace();
	}

	@Override
	public CompletionStage<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
		System.out.println("reply.");
		return null;
	}
	
	public void sendMessage(Object message) throws RemotingException {
		peer.send(message);
	}
}